package naga.x.net.proxy;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.util.IOUtils;

import naga.x.App;
import naga.x.net.ClientSessionService;
import naga.x.net.Packet;
import naga.x.net.Session;
import naga.x.net.impl.pomelo.PomeloEncoder;
import naga.x.net.impl.pomelo.PomeloPacket;
import naga.x.net.proxy.codec.ProxyPacketDecoder;
import naga.x.net.proxy.codec.ProxyPacketEncoder;
import naga.x.util.SessionUtils;
import naga.x.util.StringUtils;

/**
 * 代理服务器与游戏服连接Service
 * @author yang.li
 *
 */
public class ProxyClientSessionService extends ClientSessionService {
	
	// 游戏服与代理服务器连接的session id's key
	protected static final String SESSION_ID = ".proxySessionId";
	
	// 所有代理服务器session
	protected Map<Integer, IoSession> proxySessions = new ConcurrentHashMap<Integer, IoSession>();
	// 代理服务器session id生成器
	protected AtomicInteger sessIds = new AtomicInteger();
	
	public ProxyClientSessionService() {}
	
	@Override
	public void sessionCreated(IoSession session) throws Exception {
		Integer sessId = sessIds.incrementAndGet();
		session.setAttribute(SESSION_ID, sessId);
		proxySessions.put(sessId, session);
		LOG.info("[{}]PROXY SESSION CREATED[{}]{}", getId(), session.getRemoteAddress(), session.getId());
	}

	@Override
	public void sessionClosed(IoSession session) throws Exception {
		// 通知所有客户端断开连接
		Integer sessId = (Integer) session.getAttribute(SESSION_ID);
		for (Session sess : sessions.values()) {
			if (sess.getId() >> 32 == sessId.intValue()) {
				sess.close(true);
			}
		}
		proxySessions.remove(sessId);
		LOG.info("[{}]PROXY SESSION CLOSED[{}]{}", getId(), session.getRemoteAddress(), session.getId());
	}
	
	@Override
	public void messageReceived(IoSession session, Object message) throws Exception {
		if (message instanceof ProxyPacket) {
			ProxyPacket packet = (ProxyPacket) message;
			Session sess = getSession(session, packet.getSessionId());
			switch (packet.getOpCode()) {
				case ProxyOpCode.SESSION_CLOSE:
//					sessionClose((ProxyClientSession) sess);
					fireSessionClosed(sess);
					break;
				case ProxyOpCode.SYNC_CLIENT_IP:
					syncClientIp(sess, packet);
					break;
				default:
					packetDispatch(sess, packet);
					break;
			}
			
		}
	}
	
	/**
	 * 获取客户端session
	 * @param proxySession 游戏服与代理服的连接会话
	 * @param clientSessionId 客户端与代理服连接的session id,由代理服分配
	 * @return
	 */
	protected Session getSession(IoSession proxySession, int clientSessionId) {
		int proxySessionId = (int) proxySession.getAttribute(SESSION_ID); // 游戏服与代理服连接的session id,由游戏服分配(@see sessIds)
		Long id = ((long) proxySessionId << 32) | clientSessionId;
		Session session = sessions.get(id);
		if (session == null) {
			session = new ProxyClientSession(proxySession, App.getApp().getPacketDispatcher(), clientSessionId, this);
//			sessions.put(id, sess);
//			LOG.info("[{}]SESSION CREATED[{}]{}", getId(), sess.getRemoteAddress(), sess.getId());
			fireSessionCreated(session);
		}
		return session;
	}
	
	private void syncClientIp(Session sess, ProxyPacket packet) {
		IoBuffer buf = packet.getData();
		if (buf != null) {
			String ip = StringUtils.bytesToString(buf.array(), buf.position(), buf.remaining(), naga.x.util.IOUtils.UTF8);
			if (ip != null) {
				sess.setAttribute(SessionUtils.ADDRESS, ip);
			}
		}
	}
	
	private void packetDispatch(Session sess, ProxyPacket packet) {
		IoBuffer buf = packet.getData();
		Class<?> clazz = App.getApp().getConfig().getDictionary().getRequestClass(packet.getOpCode());
		if (clazz != null && PomeloPacket.class.isAssignableFrom(clazz)) {
			try {
				PomeloPacket req = JSON.parseObject(buf.array(), buf.position(), buf.remaining(), IOUtils.UTF8, clazz);
				req.setOpCode(packet.getOpCode());
				req.setPackageType(packet.getType());
				req.setSerial(packet.getSerial());
				sess.received(req);
			} catch (Exception e) {
				LOG.error("[" + getId() + "]PACKET PARSE ERROR[" + packet.getOpCode() + "=" + clazz, e);
			}
		} else {
			LOG.info("[{}]PACKET PARSE ERROR[{}={}]", getId(), packet.getOpCode(), clazz);
		}
	}
	
//	private void sessionClose(ProxyClientSession sess) {
//		sess.state = State.CLOSED;
//		sessions.remove(sess.getId());
//		App.getApp().getEventManager().fireEvent(Events.EVENT_SESSION_CLOSED, sess);
//		LOG.info("[{}]SESSION CLOSED[{}]{}", getId(), sess.getRemoteAddress(), sess.getId());
//	}

	@Override
	public String getId() {
		return "PROXY";
	}

	@Override
	protected IoFilter getCodecFilter() {
		return new ProtocolCodecFilter(new ProxyPacketEncoder(), new ProxyPacketDecoder());
	}

	@Override
	protected int getPort() {
//		return App.getApp().getConfig().getIntValue(Config.PROXY_PORT);
		return App.getApp().getConfig().proxyPort;
	}

//	@Override
//	public boolean update() {
////		Iterator<Entry<Long, Session>> itr = sessions.entrySet().iterator();
//		Iterator<Session> itr = sessions.values().iterator();
//		while (itr.hasNext()) {
////			Entry<Long, Session> e = itr.next();
//			Session sess = itr.next();//e.getValue();
//			if (sess != null) {
//				if (sess.isClosing()) {
//					sessionClose((ProxyClientSession) sess);
//				} else {
//					pool.execute(sess);
//				}
//			}
//		}
//		return true;
//	}

	@Override
	public void broadcast(Packet packet) {
		if (packet instanceof PomeloPacket) {
			PomeloPacket pp = (PomeloPacket) packet;
			Iterator<IoSession> proxies = proxySessions.values().iterator();
			while (proxies.hasNext()) {
				IoSession proxy = proxies.next();
				proxy.write(wrapPacket(pp));
			}
		}
	}
	
	private ProxyPacket wrapPacket(PomeloPacket packet) {
		ProxyPacket wrapped = new ProxyPacket(ProxyOpCode.BROADCAST, 128, -1, packet.getSerial(), packet.getPackageType());
		wrapped.put(PomeloEncoder.encode(packet));
		wrapped.getData().flip();
		return wrapped;
	}

}
